package com.winning.bi.amqp;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.SerializationUtils;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

/**
 * AMQP协议 发送消息工具类
 * 
 * @author 张大川
 * 
 *
 */
public class SendMessageUtils {
	private static Logger log = LoggerFactory.getLogger(SendMessageUtils.class);
	ConnectionFactory factory;
	Connection connection;

	public SendMessageUtils() throws IOException, TimeoutException {
		Configuration config = null;
		try {
			config = new PropertiesConfiguration("config/amqp.properties");
		} catch (ConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		factory = new ConnectionFactory();
		factory.setHost(config.getString("hostname"));
		factory.setUsername(config.getString("username"));
		factory.setPassword(config.getString("password"));
		factory.setConnectionTimeout(4000);
		if (config.containsKey("port")) {
			factory.setPort(config.getInt("port"));
		} else {
			factory.setPort(AMQP.PROTOCOL.PORT);
		}
		// factory.useNio();

	}

	/**
	 * 
	 * 
	 * @param queueName
	 *            队列名称
	 * @param object
	 *            要传送的对象
	 * @throws Exception
	 */
	public void sendMessage(String queueName, Object object) throws Exception {
		// Configuration config = new
		// PropertiesConfiguration("config/amqp.properties");
		// config.get

		try {
			connection = factory.newConnection();
			// connection.

			Channel channel = connection.createChannel();

			channel.queueDeclare(queueName, true, false, false, null);

			channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,
					SerializationUtils.serialize(object));

			log.info("send");

			channel.close();

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (Exception ignore) {
				}
			}
		}

	}

	public void close() throws IOException {
		connection.close();
	}
}
